package com.kafka.controller;


import com.kafka.consumer.KafkaConsumer;
import com.kafka.provider.KafkaProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;


@RestController
public class kafkaController {


    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private KafkaListenerEndpointRegistry registry;

    @Autowired
    private KafkaConsumer kafkaConsumer;

//    @Autowired
//    private AdminClient adminClient;

    @Autowired
    private KafkaProvider kafkaProvider;

    @GetMapping("/sendKafka")
    public void sendKafka(String message) throws Exception {
//        registry.getListenerContainer("task").stop();
        try {
            for (int i = 0; i < 20; i++) {
                kafkaProvider.sendChatMsg("logMessage", "老许", message);
                // kafkaProvider.sendChatMsg("ChatMsg", "老王", message);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

//    @GetMapping("/receiveKafka1")
//    public void receiveKafka1() throws Exception {
//        registry.getListenerContainer("task").start();
//        kafkaConsumer.startConsumer("老许", "task");
//    }
//
//    @GetMapping("/receiveKafka2")
//    public void receiveKafka2() throws Exception {
//        registry.getListenerContainer("task").start();
//        kafkaConsumer.startConsumer("老王", "task");
//    }


}
